package com.deep.flink;

import com.ibm.icu.impl.Row;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author sw
 * @create 2023-07-03 12:34
 */
public class Test5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        EnvironmentSettings environmentSettings = EnvironmentSettings
                .newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, environmentSettings);
        DataStreamSource<String> lines = env.socketTextStream("localhost", 9999);

        SingleOutputStreamOperator<WC> streamOperator = lines.flatMap(new FlatMapFunction<String, WC>() {
            @Override
            public void flatMap(String value, Collector<WC> out) throws Exception {
                String[] strings = value.split(" ");
                for (String string : strings) {
                    out.collect(new WC(string, 1L));
                }
            }
        });

        // 从DataStream 获取一个 Table
        Table table = tableEnv.fromDataStream(streamOperator);
        table.printSchema();

        // 查询 Table
        Table filter = table
                .select($("word"), $("count"))
                .filter($("word").isNotEqual(""));


        Table result = filter
                .groupBy($("word"))
                .select($("word"), $("count").sum().as("word_count"));

        // 输出 Table
        tableEnv.toRetractStream(result, Row.class).print();
        env.execute();
    }
}
